/*-------------------------------------------------------------------------
 *
 * hashinsert.c
 *	  在Postgres的哈希表中插入项。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/access/hash/hashinsert.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/hash.h"
#include "access/hash_xlog.h"
#include "access/xloginsert.h"
#include "miscadmin.h"
#include "storage/buf_internals.h"
#include "storage/lwlock.h"
#include "storage/predicate.h"
#include "utils/rel.h"

static void fc__hash_vacuum_one_page(Relation fc_rel, Relation fc_hrel,
								  Buffer fc_metabuf, Buffer fc_buf);

/*
 *	_hash_doinsert() -- 处理单个索引元组的插入。
 *
 *		这个例程由公共接口例程 hashbuild 和 hashinsert 调用。到此为止，itup 已经完全填充。
 */
void _hash_doinsert(Relation fc_rel, IndexTuple fc_itup, Relation fc_heapRel)
{
	Buffer		fc_buf = InvalidBuffer;
	Buffer		fc_bucket_buf;
	Buffer		fc_metabuf;
	HashMetaPage fc_metap;
	HashMetaPage fc_usedmetap = NULL;
	Page		fc_metapage;
	Page		fc_page;
	HashPageOpaque fc_pageopaque;
	Size		fc_itemsz;
	bool		fc_do_expand;
	uint32		fc_hashkey;
	Bucket		fc_bucket;
	OffsetNumber fc_itup_off;

	/*
	 * 获取项目的哈希键（存储在索引元组本身中）。
	 */
	fc_hashkey = _hash_get_indextuple_hashkey(fc_itup);

	/* 计算项目大小 */
	fc_itemsz = IndexTupleSize(fc_itup);
	fc_itemsz = MAXALIGN(fc_itemsz);	/* 为了安全，PageAddItem 将做到这一点，但我们需要保持一致 */

restart_insert:

	/*
	 * 读取元页面。我们还没有锁定它；HashMaxItemSize() 将检查 pd_pagesize_version，但它是不能改变的，因此我们可以在不加锁的情况下检查它。
	 */
	fc_metabuf = _hash_getbuf(fc_rel, HASH_METAPAGE, HASH_NOLOCK, LH_META_PAGE);
	fc_metapage = BufferGetPage(fc_metabuf);

	/*
	 * 检查项目是否可以完全适应哈希页面。（最终，如果不能，我们应该尝试应用 TOAST 方法。）注意在这一点上，itemsz 不包括 ItemId。
	 *
	 * XXX 如果我们只是存储哈希键，这段代码是没有用的。
	 */
	if (fc_itemsz > HashMaxItemSize(fc_metapage))
		ereport(ERROR,
				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
				 errmsg("index row size %zu exceeds hash maximum %zu",
						fc_itemsz, HashMaxItemSize(fc_metapage)),
				 errhint("Values larger than a buffer page cannot be indexed.")));

	/* 锁定目标桶的主桶页面。 */
	fc_buf = _hash_getbucketbuf_from_hashkey(fc_rel, fc_hashkey, HASH_WRITE,
										  &fc_usedmetap);
	Assert(fc_usedmetap != NULL);

	CheckForSerializableConflictIn(fc_rel, NULL, BufferGetBlockNumber(fc_buf));

	/* 记住主桶缓冲区以便在结束时释放它的引脚。 */
	fc_bucket_buf = fc_buf;

	fc_page = BufferGetPage(fc_buf);
	fc_pageopaque = HashPageGetOpaque(fc_page);
	fc_bucket = fc_pageopaque->hasho_bucket;

	/*
	 * 如果该桶正在被拆分的过程中，尽量在插入之前完成拆分，因为这可能会为插入腾出空间，而无需分配额外的溢出页面。只有在我们试图插入到我们正在删除元组的桶（“旧”桶）中时，完成拆分才是有意义的，而不是在我们试图插入到元组被移动到的桶（“新”桶）中时。
	 */
	if (H_BUCKET_BEING_SPLIT(fc_pageopaque) && IsBufferCleanupOK(fc_buf))
	{
		/* 在完成拆分之前释放桶缓冲区上的锁。 */
		LockBuffer(fc_buf, BUFFER_LOCK_UNLOCK);

		_hash_finish_split(fc_rel, fc_metabuf, fc_buf, fc_bucket,
						   fc_usedmetap->hashm_maxbucket,
						   fc_usedmetap->hashm_highmask,
						   fc_usedmetap->hashm_lowmask);

		/* 释放旧缓冲区和元缓冲区上的引脚。重试插入。 */
		_hash_dropbuf(fc_rel, fc_buf);
		_hash_dropbuf(fc_rel, fc_metabuf);
		goto restart_insert;
	}

	/* 执行插入 */
	while (PageGetFreeSpace(fc_page) < fc_itemsz)
	{
		BlockNumber fc_nextblkno;

		/*
		 * 检查当前页面是否有任何死亡元组。如果有，删除这些元组，并查看在移动到桶链的下一个页面之前，是否有空间可以插入新项目。
		 */
		if (H_HAS_DEAD_TUPLES(fc_pageopaque))
		{

			if (IsBufferCleanupOK(fc_buf))
			{
				fc__hash_vacuum_one_page(fc_rel, fc_heapRel, fc_metabuf, fc_buf);

				if (PageGetFreeSpace(fc_page) >= fc_itemsz)
					break;		/* 好吧，现在我们有足够的空间 */
			}
		}

		/*
		 * 此页面上没有空间；检查是否有溢出页面
		 */
		fc_nextblkno = fc_pageopaque->hasho_nextblkno;

		if (BlockNumberIsValid(fc_nextblkno))
		{
			/*
			 * 溢出页面存在；去获取它。如果没有空间，我们将在上面的循环测试中发现。我们总是释放锁和引脚，如果这是一个溢出页面，但如果这是主桶页面，则只释放锁，因为主桶的引脚必须在整个扫描过程中保持。
			 */
			if (fc_buf != fc_bucket_buf)
				_hash_relbuf(fc_rel, fc_buf);
			else
				LockBuffer(fc_buf, BUFFER_LOCK_UNLOCK);
			fc_buf = _hash_getbuf(fc_rel, fc_nextblkno, HASH_WRITE, LH_OVERFLOW_PAGE);
			fc_page = BufferGetPage(fc_buf);
		}
		else
		{
			/*
			 * 我们在桶链的末尾，没有找到有足够空间的页面。分配一个新的溢出页面。
			 */

			/* 在不修改缓冲区的情况下释放我们的写锁 */
			LockBuffer(fc_buf, BUFFER_LOCK_UNLOCK);

			/* 链接到新的溢出页面 */
			fc_buf = _hash_addovflpage(fc_rel, fc_metabuf, fc_buf, (fc_buf == fc_bucket_buf));
			fc_page = BufferGetPage(fc_buf);

			/* 现在应该适合，鉴于上述测试 */
			Assert(PageGetFreeSpace(fc_page) >= fc_itemsz);
		}
		fc_pageopaque = HashPageGetOpaque(fc_page);
		Assert((fc_pageopaque->hasho_flag & LH_PAGE_TYPE) == LH_OVERFLOW_PAGE);
		Assert(fc_pageopaque->hasho_bucket == fc_bucket);
	}

	/*
	 * 写锁定元页面，以便我们可以递增元组计数。在递增后，检查是否是拆分的时机。
	 */
	LockBuffer(fc_metabuf, BUFFER_LOCK_EXCLUSIVE);

	/* 执行更新。在更改被记录之前不进行 ereport(ERROR) */
	START_CRIT_SECTION();

	/* 找到有足够空间的页面，因此在此添加项目 */
	fc_itup_off = _hash_pgaddtup(fc_rel, fc_buf, fc_itemsz, fc_itup);
	MarkBufferDirty(fc_buf);

	/* 元页面操作 */
	fc_metap = HashPageGetMeta(fc_metapage);
	fc_metap->hashm_ntuples += 1;

	/* 确保这与 _hash_expandtable() 保持同步 */
	fc_do_expand = fc_metap->hashm_ntuples >
		(double) fc_metap->hashm_ffactor * (fc_metap->hashm_maxbucket + 1);

	MarkBufferDirty(fc_metabuf);

	/* XLOG 相关内容 */
	if (RelationNeedsWAL(fc_rel))
	{
		xl_hash_insert fc_xlrec;
		XLogRecPtr	fc_recptr;

		fc_xlrec.offnum = fc_itup_off;

		XLogBeginInsert();
		XLogRegisterData((char *) &fc_xlrec, SizeOfHashInsert);

		XLogRegisterBuffer(1, fc_metabuf, REGBUF_STANDARD);

		XLogRegisterBuffer(0, fc_buf, REGBUF_STANDARD);
		XLogRegisterBufData(0, (char *) fc_itup, IndexTupleSize(fc_itup));

		fc_recptr = XLogInsert(RM_HASH_ID, XLOG_HASH_INSERT);

		PageSetLSN(BufferGetPage(fc_buf), fc_recptr);
		PageSetLSN(BufferGetPage(fc_metabuf), fc_recptr);
	}

	END_CRIT_SECTION();

	/* 释放元页面上的锁，但保持固定 */
	LockBuffer(fc_metabuf, BUFFER_LOCK_UNLOCK);

	/*
	 * 释放修改过的页面，并确保释放主页面上的固定
	 * 页面。
	 */
	_hash_relbuf(fc_rel, fc_buf);
	if (fc_buf != fc_bucket_buf)
		_hash_dropbuf(fc_rel, fc_bucket_buf);

	/* 如果需要分裂，则尝试进行分裂 */
	if (fc_do_expand)
		_hash_expandtable(fc_rel, fc_metabuf);

	/* 最后释放我们的元页面上的固定 */
	_hash_dropbuf(fc_rel, fc_metabuf);
}

/*
 *	_hash_pgaddtup() -- 将元组添加到索引中的特定页面。
 *
 * 此例程将元组添加到请求的页面；它不会写出
 * 页面。 在没有固定和写锁定
 * 目标缓冲区的情况下调用此函数是错误的。
 *
 * 返回插入元组的偏移量。此函数
 * 负责保持哈希索引
 * 页面中元组按哈希键值排序的条件。
 */
OffsetNumber _hash_pgaddtup(Relation fc_rel, Buffer fc_buf, Size fc_itemsize, IndexTuple fc_itup)
{
	OffsetNumber fc_itup_off;
	Page		fc_page;
	uint32		fc_hashkey;

	_hash_checkpage(fc_rel, fc_buf, LH_BUCKET_PAGE | LH_OVERFLOW_PAGE);
	fc_page = BufferGetPage(fc_buf);

	/* 找到插入元组的位置（保持页面的哈希键排序） */
	fc_hashkey = _hash_get_indextuple_hashkey(fc_itup);
	fc_itup_off = _hash_binsearch(fc_page, fc_hashkey);

	if (PageAddItem(fc_page, (Item) fc_itup, fc_itemsize, fc_itup_off, false, false)
		== InvalidOffsetNumber)
		elog(ERROR, "failed to add index item to \"%s\"",
			 RelationGetRelationName(fc_rel));

	return fc_itup_off;
}

/*
 *	_hash_pgaddmultitup() -- 将元组向量添加到
 *							 索引中的特定页面。
 *
 * 此例程在锁定和元组排序方面与
 * _hash_pgaddtup() 有相同的要求。
 *
 * 返回插入元组的偏移量数组。
 */
void _hash_pgaddmultitup(Relation fc_rel, Buffer fc_buf, IndexTuple *fc_itups,
					OffsetNumber *fc_itup_offsets, uint16 fc_nitups)
{
	OffsetNumber fc_itup_off;
	Page		fc_page;
	uint32		fc_hashkey;
	int			fc_i;

	_hash_checkpage(fc_rel, fc_buf, LH_BUCKET_PAGE | LH_OVERFLOW_PAGE);
	fc_page = BufferGetPage(fc_buf);

	for (fc_i = 0; fc_i < fc_nitups; fc_i++)
	{
		Size		fc_itemsize;

		fc_itemsize = IndexTupleSize(fc_itups[fc_i]);
		fc_itemsize = MAXALIGN(fc_itemsize);

		/* 找到插入元组的位置（保持页面的哈希键排序） */
		fc_hashkey = _hash_get_indextuple_hashkey(fc_itups[fc_i]);
		fc_itup_off = _hash_binsearch(fc_page, fc_hashkey);

		fc_itup_offsets[fc_i] = fc_itup_off;

		if (PageAddItem(fc_page, (Item) fc_itups[fc_i], fc_itemsize, fc_itup_off, false, false)
			== InvalidOffsetNumber)
			elog(ERROR, "failed to add index item to \"%s\"",
				 RelationGetRelationName(fc_rel));
	}
}

/*
 * _hash_vacuum_one_page - 清理仅一个索引页面。
 *
 * 尝试从给定页面中删除 LP_DEAD 项。我们必须在调用此函数之前对正在修改的页面获取清理
 * 锁。
 */

static void fc__hash_vacuum_one_page(Relation fc_rel, Relation fc_hrel, Buffer fc_metabuf, Buffer fc_buf)
{
	OffsetNumber fc_deletable[MaxOffsetNumber];
	int			fc_ndeletable = 0;
	OffsetNumber fc_offnum,
				fc_maxoff;
	Page		fc_page = BufferGetPage(fc_buf);
	HashPageOpaque fc_pageopaque;
	HashMetaPage fc_metap;

	/* 扫描页面中的每个元组以查看其是否标记为 LP_DEAD */
	fc_maxoff = PageGetMaxOffsetNumber(fc_page);
	for (fc_offnum = FirstOffsetNumber;
		 fc_offnum <= fc_maxoff;
		 fc_offnum = OffsetNumberNext(fc_offnum))
	{
		ItemId		fc_itemId = PageGetItemId(fc_page, fc_offnum);

		if (ItemIdIsDead(fc_itemId))
			fc_deletable[fc_ndeletable++] = fc_offnum;
	}

	if (fc_ndeletable > 0)
	{
		TransactionId fc_latestRemovedXid;

		fc_latestRemovedXid =
			index_compute_xid_horizon_for_tuples(fc_rel, fc_hrel, fc_buf,
												 fc_deletable, fc_ndeletable);

		/*
		 * 写锁定元页面，以便我们可以减少元组计数。
		 */
		LockBuffer(fc_metabuf, BUFFER_LOCK_EXCLUSIVE);

		/* 直到更改被记录，不要ereport(ERROR) */
		START_CRIT_SECTION();

		PageIndexMultiDelete(fc_page, fc_deletable, fc_ndeletable);

		/*
		 * 将页面标记为不包含任何 LP_DEAD 项。这并不
		 * 一定是正确的（可能有一些最近被标记，
		 * 但不在我们的目标项列表中），但这几乎
		 * 总是正确的，并且似乎不值得额外的页面扫描来
		 * 检查它。请记住，LH_PAGE_HAS_DEAD_TUPLES 反正只是一个提示。
		 */
		fc_pageopaque = HashPageGetOpaque(fc_page);
		fc_pageopaque->hasho_flag &= ~LH_PAGE_HAS_DEAD_TUPLES;

		fc_metap = HashPageGetMeta(BufferGetPage(fc_metabuf));
		fc_metap->hashm_ntuples -= fc_ndeletable;

		MarkBufferDirty(fc_buf);
		MarkBufferDirty(fc_metabuf);

		/* XLOG 相关内容 */
		if (RelationNeedsWAL(fc_rel))
		{
			xl_hash_vacuum_one_page fc_xlrec;
			XLogRecPtr	fc_recptr;

			fc_xlrec.latestRemovedXid = fc_latestRemovedXid;
			fc_xlrec.ntuples = fc_ndeletable;

			XLogBeginInsert();
			XLogRegisterBuffer(0, fc_buf, REGBUF_STANDARD);
			XLogRegisterData((char *) &fc_xlrec, SizeOfHashVacuumOnePage);

			/*
			 * 无论我们是否存储整个缓冲区，我们都需要目标偏移量数组，
			 * 以便我们能够在备用服务器上找到 latestRemovedXid。
			 */
			XLogRegisterData((char *) fc_deletable,
							 fc_ndeletable * sizeof(OffsetNumber));

			XLogRegisterBuffer(1, fc_metabuf, REGBUF_STANDARD);

			fc_recptr = XLogInsert(RM_HASH_ID, XLOG_HASH_VACUUM_ONE_PAGE);

			PageSetLSN(BufferGetPage(fc_buf), fc_recptr);
			PageSetLSN(BufferGetPage(fc_metabuf), fc_recptr);
		}

		END_CRIT_SECTION();

		/*
		 * 释放元页面上的写锁，因为我们已更新元组
		 * 计数。
		 */
		LockBuffer(fc_metabuf, BUFFER_LOCK_UNLOCK);
	}
}
